package net.bwie.realtime.vehicle.dwd.job.one;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.utils.DorisUtil;
import net.bwie.realtime.utils.KafkaUtil;
import net.bwie.realtime.vehicle.dwd.bean.DangerousDriver;
import net.bwie.realtime.vehicle.dwd.bean.Monitor;
import net.bwie.realtime.vehicle.dwd.utils.OverSpeedUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @author ASUS
 *
 */
/*
 * TODO 危险驾驶分析
 *  判定条件：机动车在2分钟内，超速通过卡口超过3次以上且每次超速的超过了规定速度的20%以上;这样的机动车涉嫌危险驾驶
 *  没有限速的路口设定默认限速60
 */

public class DangerousDriverController {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 每 1000ms 开始一次 checkpoint
        env.enableCheckpointing(5000);

        // 获取 kafka 数据源
        DataStream<String> ds0 = KafkaUtil.consumerKafka(env, "dwd_speeding_info");

        // 数据处理
        DataStream<String> ds1 = trafficProcessJson(ds0);

//        ds0.print();

        // 转换数据结构为实体类
        SingleOutputStreamOperator<Monitor> ds2 = ds1.map(x -> JSON.parseObject(x, Monitor.class));

//        ds2.print();

        // 水位线  允许数据迟到五秒
        SingleOutputStreamOperator<Monitor> ds3 = ds2.assignTimestampsAndWatermarks(WatermarkStrategy
                .<Monitor>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<Monitor>() {
                    @Override
                    public long extractTimestamp(Monitor Monitor, long l) {
                        return Monitor.getActionTime();
                    }
                }));


        ds3.print();


        // 定义Pattern
        Pattern<Monitor, Monitor> pattern = Pattern.<Monitor>begin("first", AfterMatchSkipStrategy.skipPastLastEvent())
                .where(new SimpleCondition<Monitor>() {
                    @Override
                    public boolean filter(Monitor Monitor) throws Exception {
                        int limitSpeed = Monitor.getSpeedLimit();
                        int speed = Monitor.getSpeed();
                        return speed > limitSpeed * 1.2;
                    }
                })
                // TODO 匹配3次满足条件的事件
                .times(3)
                // TODO 满足2分钟内条件
                .within(Time.minutes(2));

        // CEP
        PatternStream<Monitor> patternStream = CEP.pattern(ds3.keyBy(Monitor::getCar), pattern);


        DataStream<DangerousDriver> ds4 = patternStream.select(new PatternSelectFunction<Monitor, DangerousDriver>() {
            @Override
            public DangerousDriver select(Map<String, List<Monitor>> map) throws Exception {
                Monitor first = map.get("first").get(0);
                return new DangerousDriver(
                        first.getActionTime().toString(),
                        first.getActionTime().toString(),
                        first.getMonitorId(),
                        first.getCameraId(),
                        first.getCar(),
                        first.getSpeed(),
                        first.getSpeedLimit()
                );
            }
        });

        ds4.print("4:");

        DataStream<DangerousDriver> ds5 = ds4.keyBy(DangerousDriver::getCar)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<DangerousDriver>() {
                    @Override
                    public DangerousDriver reduce(DangerousDriver dangerousDriver, DangerousDriver t1) throws Exception {
                        return new DangerousDriver(
                                dangerousDriver.getStartTime(),
                                t1.getEndTime(),
                                dangerousDriver.getMonitorId(),
                                dangerousDriver.getCameraId(),
                                dangerousDriver.getCar(),
                                dangerousDriver.getSpeed(),
                                dangerousDriver.getSpeedLimit()
                        );
                    }
                });
        // 转换数据格式
        DataStream<String> ds6 = ds5.map(new MapFunction<DangerousDriver, String>() {
            @Override
            public String map(DangerousDriver dangerousDriver) throws Exception {
                return JSON.toJSONString(dangerousDriver);
            }
        });


        ds6.print();
        // 写入Doris数据库
        DorisUtil.saveToDoris(ds6, "vehicle_realtime_report", "vehicle_break_track_report");

        env.execute("DangerousDriverController");
    }

    private static DataStream<String> trafficProcessJson(DataStream<String> ds1) {
        return ds1.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String monitorId = jsonObject.getString("monitorId");

                // 从MonitorUtil的缓存中获取限速信息，而不是重新加载
                // 注意：需要在MonitorUtil中添加一个获取缓存的方法
                Integer speedLimit = OverSpeedUtil.getSpeedLimitFromCache(monitorId);


                // 如果缓存中没有，使用默认值（例如60）
                if (speedLimit == null) {
                    speedLimit = 60;
                    // 可选：如果需要，可以在这里触发一次数据库查询加载该monitorId的信息
                    OverSpeedUtil.preloadAreaInfo(monitorId);
                }

                // 将获取到的限速信息放入JSON对象
                jsonObject.put("speedLimit", speedLimit);

                return jsonObject.toJSONString();
            }
        });
    }
}
